package com.deep.flink.hudi;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.util.DataFormatConverters;
import org.apache.flink.table.types.logical.IntType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.utils.TypeConversions;
import org.apache.flink.types.Row;

import java.util.*;


public class SyncCustomer {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        String inputTableSQL = "create table tbl_customer_mysql (\n" +
                "  id STRING PRIMARY KEY NOT ENFORCED,\n" +
                "  customer_relationship_id STRING,\n" +
                "  create_date_time STRING,\n" +
                "  update_date_time STRING,\n" +
                "  deleted STRING,\n" +
                "  name STRING,\n" +
                "  idcard STRING,\n" +
                "  birth_year STRING,\n" +
                "  gender STRING,\n" +
                "  phone STRING,\n" +
                "  wechat STRING,\n" +
                "  qq STRING,\n" +
                "  email STRING,\n" +
                "  area STRING,\n" +
                "  leave_school_date STRING,\n" +
                "  graduation_date STRING,\n" +
                "  bxg_student_id STRING,\n" +
                "  creator STRING,\n" +
                "  origin_type STRING,\n" +
                "  origin_channel STRING,\n" +
                "  tenant STRING,\n" +
                "  md_id STRING\n" +
                ")WITH (\n" +
                "  'connector' = 'mysql-cdc',\n" +
                "  'hostname' = 'localhost',\n" +
                "  'port' = '3320',\n" +
                "  'username' = 'root',\n" +
                "  'password' = '123456',\n" +
                "  'server-time-zone' = 'Asia/Shanghai',\n" +
                "  'debezium.snapshot.mode' = 'initial',\n" +
                "  'database-name' = 'itcast_nev',\n" +
                "  'table-name' = 'customer'\n" +
                ") ";

        tableEnv.executeSql(inputTableSQL);


        String outputTableSQL = "CREATE TABLE edu_customer_hudi(\n" +
                "  id STRING PRIMARY KEY NOT ENFORCED,\n" +
                "  customer_relationship_id STRING,\n" +
                "  create_date_time STRING,\n" +
                "  update_date_time STRING,\n" +
                "  deleted STRING,\n" +
                "  name STRING,\n" +
                "  idcard STRING,\n" +
                "  birth_year STRING,\n" +
                "  gender STRING,\n" +
                "  phone STRING,\n" +
                "  wechat STRING,\n" +
                "  qq STRING,\n" +
                "  email STRING,\n" +
                "  area STRING,\n" +
                "  leave_school_date STRING,\n" +
                "  graduation_date STRING,\n" +
                "  bxg_student_id STRING,\n" +
                "  creator STRING,\n" +
                "  origin_type STRING,\n" +
                "  origin_channel STRING,\n" +
                "  tenant STRING,\n" +
                "  md_id STRING,\n" +
                "  part STRING\n" +
                ")\n" +
                "PARTITIONED BY (part)\n" +
                "WITH(\n" +
                "  'connector'='hudi',\n" +
                "  'path'= 'hdfs://localhost:9000/user/deeprado/hudi-warehouse/edu_customer_hudi', \n" +
                "  'table.type'= 'MERGE_ON_READ',\n" +
                "  'hoodie.datasource.write.recordkey.field'= 'id', \n" +
                "  'write.precombine.field'= 'create_date_time',\n" +
                "  'write.tasks'= '1',\n" +
                "  'read.tasks'= '1',\n" +
                "  'write.rate.limit'= '2000', \n" +
                "  'compaction.tasks'= '1', \n" +
                "  'compaction.async.enabled'= 'true',\n" +
                "  'compaction.trigger.strategy'= 'num_commits',\n" +
                "  'compaction.delta_commits'= '1',\n" +
                "  'changelog.enabled'= 'true'\n" +
                ") ";

        tableEnv.executeSql(outputTableSQL);

        String syncSQL= "insert into edu_customer_hudi \n" +
                "select *, CAST(CURRENT_DATE AS STRING) AS part from tbl_customer_mysql ";

        tableEnv.executeSql(syncSQL);

         env.execute("Api_Sink");
    }
}
